package com.cmsr.hdpf.source

import com.cmsr.hdpf.util.ConfigHelper
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer

import java.util.Properties

object KafkaSource {
    
    def GetKafkaConsumer(data: String, group: String): FlinkKafkaConsumer[String] = {
        
        
        val props: Properties = new Properties()
        props.setProperty("bootstrap.servers", ConfigHelper.parse_source_kafka_bootstrapServers)
        props.setProperty("group.id", group + s"-${System.currentTimeMillis}") // group.id 避免重名？单线程还是多线程？
        println("kafka group: " + group + s"-${System.currentTimeMillis}")
        props.setProperty("enable.auto.commit", ConfigHelper.parse_source_kafka_enableAutoCommit)
        props.setProperty("auto.commit.interval.ms", ConfigHelper.parse_source_kafka_autoCommitIntervalMs)
        props.setProperty("auto.offset.reset", ConfigHelper.parse_source_kafka_autoOffsetReset)
        props.setProperty("request.timeout.ms", "60000") // 请求延迟时间
        /*props.setProperty("client.id", scala.util.Random.nextInt(9999) + s"-${System.currentTimeMillis}")*/
        
        val consumer = new FlinkKafkaConsumer[String](
            data, // 目标topic
            new SimpleStringSchema(),
            props
        )
        
        //consumer.setStartFromEarliest() // 从最早记录开始消费
        consumer.setStartFromLatest() // 从最新记录开始消费

        // 有些分区的时间戳不满足？
        // 读取历史数据时，新数据的到来与历史数据冲突？
//        consumer.setStartFromTimestamp(1666166400000L) // 从指定时间开始消费 数据读取的起始时间设置为整分时刻，保证滚动窗口按分钟划分！

        println(consumer)

        consumer
    }
}